package com.jwds.digital.bios.business.integration.consumer;

import cn.hutool.core.util.EnumUtil;
import cn.hutool.json.JSONUtil;
import com.jwds.digital.bios.business.core.enums.ExternalActionEnum;
import com.jwds.digital.bios.business.core.infrastructure.dto.external.ExternalSystemReceiveData;
import com.jwds.digital.bios.business.core.infrastructure.dto.feign.BasicReceiveLogReqDTO;
import com.jwds.digital.bios.business.integration.factory.ExternalDataConsumerFactory;
import com.jwds.digital.bios.business.integration.service.ExternalDataProcess;
import com.jwds.digital.bios.business.rpc.feign.BasicReceiveLogFeign;
import com.jwds.tdc.framework.common.exception.BusinessException;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.ObjectUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.streams.kstream.KStream;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Set;
import java.util.function.Consumer;

/**
 * @author hesong
 */
@Slf4j
@Component
public class ExternalDataConsumer {

    @Autowired
    private ExternalDataConsumerFactory consumerFactory;

    @Autowired
    private BasicReceiveLogFeign receiveLogFeign;

    @Bean
    public Consumer<KStream<Object, String>> externalDataProcess() {
        return procedure -> procedure.foreach((k, v) -> {
            try {
                //格式化
                List<ExternalSystemReceiveData> messageList = JSONUtil.parseArray(v).toList(ExternalSystemReceiveData.class);
                //消息过滤后消费
                Set<String> deviceMessageCodes = consumerFactory.getDeviceMessageCodes();
                messageList.stream().filter(o -> deviceMessageCodes.contains(o.getTableName())).forEach(receiveData -> {
                    oneMessageProcess(receiveData);
                });
            } catch (Exception e) {
                log.error("接收到平台消息异常：{},消息内容：{}", e, v);
            }
        });
    }

    /**
     * 消息校验处理
     *
     * @param receiveData 消息实体
     */
    private void oneMessageProcess(ExternalSystemReceiveData receiveData) {
        String tableName = receiveData.getTableName();
        BasicReceiveLogReqDTO saveDTO = new BasicReceiveLogReqDTO();
        try {
            saveDTO.setUniqueKey(receiveData.getTid());
            Object dataObj = receiveData.getData();
            saveDTO.setMessageBody(JSONUtil.toJsonStr(receiveData));
            saveDTO.setMessageCode(tableName);

            if (StringUtils.isEmpty(tableName)) {
                throw new BusinessException("500", "电文号缺失");
            }
            if (ObjectUtils.isEmpty(dataObj)) {
                throw new BusinessException("500", "电文内容缺失");
            }
            String operate = receiveData.getOperate();
            if (StringUtils.isEmpty(operate)) {
                throw new BusinessException("500", "电文操作类型缺失");
            }
            ExternalDataProcess handlerInstance = consumerFactory.getHandlerInstance(tableName);

            ExternalActionEnum externalActionEnum = EnumUtil.likeValueOf(ExternalActionEnum.class, operate);

            switch (externalActionEnum) {
                case UPDATE:
                    handlerInstance.updateProcess(dataObj);
                    break;
                case ADD:
                    handlerInstance.addProcess(dataObj);
                    break;
                case DELETE:
                    handlerInstance.deleteProcess(dataObj);
                    break;
                default:
                    break;
            }
            saveDTO.setState(1);
        } catch (Exception e) {
            String errorMsg = e.getMessage();
            saveDTO.setState(0);
            saveDTO.setErrorMsg(errorMsg);
            log.error("消息处理异常：{}", errorMsg);
        } finally {
            //保存历史数据日志
            receiveLogFeign.add(saveDTO);
        }
    }
}
